package io.kiki.sba.registry.client.task;

import io.kiki.sba.registry.client.api.ClientConfig;
import io.kiki.sba.registry.client.api.Register;
import io.kiki.sba.registry.client.provider.AbstractInternalRegister;
import io.kiki.sba.registry.client.provider.AbstractInternalRegister.SyncTask;
import io.kiki.sba.registry.client.provider.RegisterCache;
import io.kiki.sba.registry.client.remoting.Client;
import io.kiki.sba.registry.core.model.RegisterResponse;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

@Setter
@Getter
public class WorkerThread extends AbstractWorkerThread {
    private static final Logger logger = LoggerFactory.getLogger(WorkerThread.class);


    protected final TaskEventQueue requestQueue = new TaskEventQueue();

    private ClientConfig clientConfig;

    private RegisterCache registerCache;

    private AtomicBoolean inited = new AtomicBoolean(false);


    public WorkerThread(Client client, ClientConfig clientConfig, RegisterCache registerCache) {
        super(client);
        this.clientConfig = clientConfig;
        this.registerCache = registerCache;
        this.setName("RegistryWorkerThread");
        this.setDaemon(true);
    }


    @Override
    public void schedule(TaskEvent taskEvent) {
        if (inited.compareAndSet(false, true)) {
            this.start();
        }
        requestQueue.put(taskEvent);
        signal();
    }

    @Override
    public void schedule(List<TaskEvent> taskEventList) {
        if (inited.compareAndSet(false, true)) {
            this.start();
        }
        requestQueue.putAll(taskEventList);
    }


    @Override
    public void handle() {
        //noinspection InfiniteLoopStatement
        while (true) {
            try {
                // check connection status, try to reconnect to the server when connection lose
                client.ensureConnected();

                if (requestQueue.isEmpty()) {
                    await(clientConfig.getRecheckInterval());
                    continue;
                }

                Iterator<TaskEvent> iterator = requestQueue.iterator();

                while (iterator.hasNext()) {
                    client.ensureConnected();
                    TaskEvent taskEvent = iterator.next();
                    iterator.remove();
                    int sendCount = taskEvent.incSendCount();

                    // Resent needs delay when task event is not the first time to send.
                    if (sendCount != 0 && taskEvent.delayTime() > 0) {
                        continue;
                    }

                    handleTask(taskEvent);
                }

                // Cleaning completed task, it will take more time when the registration number is large.
                requestQueue.cleanCompletedTasks();
            } catch (Throwable e) {
                logger.error("[send] handle data error!", e);
            }
        }
    }

    private void handleTask(TaskEvent taskEvent) {
        if (null == taskEvent) {
            return;
        }

        try {
            taskEvent.setTriggerTime(System.currentTimeMillis());
            Register register = taskEvent.getSource();

            if (!(register instanceof AbstractInternalRegister)) {
                logger.warn("[register] register type unknown, {}", register);
                return;
            }

            AbstractInternalRegister abstractInternalRegister = (AbstractInternalRegister) register;

            SyncTask syncTask = abstractInternalRegister.assemblySyncTask();
            String requestId = syncTask.getRequestId();

            if (syncTask.isDone()) {
                logger.info("[register] register already sync succeeded, {}", register);
                return;
            }

            Object request = syncTask.getRequest();

            Object result = client.invokeSync(request);

            if (!(result instanceof RegisterResponse)) {
                logger.warn("[register] result type is wrong, {}", result);
                return;
            }

            RegisterResponse registerResponse = (RegisterResponse) result;
            if (!registerResponse.isSuccess()) {
                logger.info("[register] register to server failed, {}, {}", request, registerResponse);
                return;
            }

            boolean syncOK = abstractInternalRegister.syncOK(requestId, registerResponse.getVersion(), registerResponse.isRefused());
            if (!syncOK) {
                logger.info("[register] requestId has expired, ignore this response, {}, {}, {}", requestId, request, registerResponse);
                return;
            }

            if (!register.isEnabled()) {
                registerCache.remove(register.getRegistId());
            }

            if (registerResponse.isRefused()) {
                logger.info("[register] register refused by server, {}, {}, {}", requestId, request, registerResponse);
            } else {
                logger.info("[register] register to server success, {}, {}, {}", requestId, request, registerResponse);
            }
        } catch (Exception e) {
            logger.error("[send] handle request failed, {}", taskEvent, e);
        }
    }
}
